package com.h.publisher;

import com.h.util.RabbitMQConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.UUID;

/**
 * @author: hlj
 * @date: 2022年05月30日 15:34
 */
public class Publisher {
    public static final String QUEUE_NAME = "H";
    public static final String WORK_QUEUE_NAME = "WORK";
    public static final String PUBSUB_EXCHANGE_NAME = "PUBSUB";
    public static final String ROUTING_EXCHANGE_NAME = "Routing";
    public static final String TOPIC_EXCHANGE_NAME = "Topic";
    public static final String PUBSUB_QUEUE_NAME1 = "PUBSUB1";
    public static final String PUBSUB_QUEUE_NAME2 = "PUBSUB2";
    public static final String ROUTING_QUEUE_NAME1 = "ROUTING1";
    public static final String ROUTING_QUEUE_NAME2 = "ROUTING2";
    public static final String TOPIC_QUEUE_NAME1 = "TOPIC1";
    public static final String TOPIC_QUEUE_NAME2 = "TOPIC2";
    public static final String RPC_QUEUE_PUBLISHER = "PUBLISHER";
    public static final String RPC_QUEUE_CONSUMER = "CONSUMER";

    //hello word模式
    @Test
    public void publisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 具体的消息
        String msg = "ZGL";
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        System.out.println("消息已发送");
        connection.close();
    }

    //WorkQueues模式
    @Test
    public void workQueuesPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(WORK_QUEUE_NAME, false, false, false, null);
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 第四个具体的消息
        for (int j = 0; j < 10; j++) {
            String msg = "WORK_QUEUE_NAME" + j;
            channel.basicPublish("", WORK_QUEUE_NAME, null, msg.getBytes());
        }
        System.out.println("消息已发送");
        connection.close();
    }

    //Pubsub模式
    @Test
    public void PubsubPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建交换机 第一个交换机名称 第二个交换机类型 Pubsub模式(FANOUT类型)
        channel.exchangeDeclare(PUBSUB_EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(PUBSUB_QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(PUBSUB_QUEUE_NAME2, false, false, false, null);
        //交换机和队列绑定 队列名 交换机名 路由规则
        channel.queueBind(PUBSUB_QUEUE_NAME1, PUBSUB_EXCHANGE_NAME, "");
        channel.queueBind(PUBSUB_QUEUE_NAME2, PUBSUB_EXCHANGE_NAME, "");
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 第四个具体的消息
        String msg = "PUBSUB模式";
        channel.basicPublish(PUBSUB_EXCHANGE_NAME, "", null, msg.getBytes());
        System.out.println("消息已发送");
        connection.close();
    }

    //Routing模式
    @Test
    public void RoutingPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建交换机 第一个交换机名称 第二个交换机类型 Routing模式(DIRECT类型)
        channel.exchangeDeclare(ROUTING_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(ROUTING_QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(ROUTING_QUEUE_NAME2, false, false, false, null);
        //交换机和队列绑定 队列名 交换机名 路由规则
        channel.queueBind(ROUTING_QUEUE_NAME1, ROUTING_EXCHANGE_NAME, "black");
        channel.queueBind(ROUTING_QUEUE_NAME2, ROUTING_EXCHANGE_NAME, "green");
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 第四个具体的消息
        channel.basicPublish(ROUTING_EXCHANGE_NAME, "black", null, "黑色的大猩猩".getBytes());
        channel.basicPublish(ROUTING_EXCHANGE_NAME, "green", null, "绿色的帽子".getBytes());
        channel.basicPublish(ROUTING_EXCHANGE_NAME, "white", null, "白色的衣服".getBytes());
        System.out.println("消息已发送");
        connection.close();
    }

    //Topic模式
    @Test
    public void TopicPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建交换机 第一个交换机名称 第二个交换机类型 Routing模式(DIRECT类型)
        channel.exchangeDeclare(TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(TOPIC_QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(TOPIC_QUEUE_NAME2, false, false, false, null);
        //交换机和队列绑定 队列名 交换机名 路由规则
        // TOPIC类型的交换机在和队列绑定时，需要以aaa.bbb.ccc.方式编写routingKey
        // 其中有两个特殊字符：*（相当于占位符），#（相当通配符）
        channel.queueBind(TOPIC_QUEUE_NAME1, TOPIC_EXCHANGE_NAME, "*.black.*");
        channel.queueBind(TOPIC_QUEUE_NAME2, TOPIC_EXCHANGE_NAME, "*.*.green");
        channel.queueBind(TOPIC_QUEUE_NAME2, TOPIC_EXCHANGE_NAME, "white.#");
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 第四个具体的消息
        channel.basicPublish(TOPIC_EXCHANGE_NAME, "big.black.green", null, "黑色的大猩猩".getBytes());
        channel.basicPublish(TOPIC_EXCHANGE_NAME, "small.cap.green", null, "绿色的帽子".getBytes());
        channel.basicPublish(TOPIC_EXCHANGE_NAME, "white.clothes.clothes.clothes", null, "白色的衣服".getBytes());
        System.out.println("消息已发送");
        connection.close();
    }

    //RPC模式
    @Test
    public void RpcPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        final Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare(RPC_QUEUE_PUBLISHER, false, false, false, null);
        channel.queueDeclare(RPC_QUEUE_CONSUMER, false, false, false, null);
        //replyTo属性告诉服务端将回复的消息放在那个队列中 correlationId 消息的唯一id
        final String uuid = UUID.randomUUID().toString();
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                .replyTo(RPC_QUEUE_CONSUMER)
                .correlationId(uuid)
                .build();
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 具体的消息
        String msg = "hello Rpc";
        channel.basicPublish("", RPC_QUEUE_PUBLISHER, properties, msg.getBytes());
        //监听队列 第一个 队列名 第二个ACK 拿到消息之后会给mq一个ack告诉mq已经拿到了true自动开启 false关闭需要手动告诉mq 第三个拿到消息之后会执行方法
        channel.basicConsume(RPC_QUEUE_CONSUMER, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String correlationId = properties.getCorrelationId();
                if (correlationId != null && correlationId.equalsIgnoreCase(uuid)) {
                    System.out.println("服务端消息:" + new String(body, "UTF-8"));
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        System.out.println("消息已发送");
        System.in.read();
        connection.close();
    }

    //确保消息的可靠性
    @Test
    public void ConfirmPublisher() throws Exception {
        //获取链接
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //创建channel管道
        Channel channel = connection.createChannel();
        //创建队列 第一个参数队列名称 第二个是否持久化只持久化队列不包括消息 第三个 是否排外 如果为TRUE 只允许一个消费者消费 第四个 当长时间没有被使用的话是否需要删除 第五个其他参数传map 可在可视化页面看到
        channel.queueDeclare("Confirm", true, false, false, null);
        //开启confirm机制 确保消息到达exchange
        channel.confirmSelect();
        //三种方式 使用异步回调机制
        channel.addConfirmListener(new ConfirmListener() {
            //消息已经到达exchange的回调函数
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息已经到达exchange");
            }

            //消息没有发送到exchange 可以做二次操作 重试或者保存到数据库
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息没有到达exchange");
            }
        });
        //设置Return回调，确认消息是否路由到了Queue
        channel.addReturnListener(new ReturnListener() {
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("只有消息没送达队列中才会执行本方法");
            }
        });
        //开启队列中消息的持久性 deliveryMode 1为不开启 2为开启
        AMQP.BasicProperties prop = new AMQP.BasicProperties
                .Builder()
                .deliveryMode(2)
                .build();
        //发布消息 第一个参数交换机的名称 ""默认交换机 第二个路由规则 默认交换机的时候为队列名称 第三个消息其他参数 第四个具体的消息
        String msg = "确保消息的可靠性";
        //mandatory true开启return机制 false不开启
        channel.basicPublish("", "Confirm", true, prop, msg.getBytes());
        System.out.println("消息已发送");
        System.in.read();
        connection.close();
        //确保消息被消费  ACK 开启手动ack
    }

}
